package com.bigdata.hadoop.hdfs;

import java.io.FileOutputStream;
import java.io.IOException;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * hdfs读取文件的流程
 * <p>
 * 整个过程可以概括为：客户端从NameNode获取文件的元数据，找到组成文件的所有数据块及其位置，然后直接与相应的DataNode建立连接，并行读取数据块，最后在本地组装成完整的文件。
 * 以下是分步详解：
 * 第1步：客户端发起打开文件请求
 * ● 客户端调用 DistributedFileSystem 对象的 open() 方法，请求打开HDFS上的一个文件。
 * ● DistributedFileSystem 通过 RPC（远程过程调用） 连接到 NameNode，获取文件的元数据信息。
 * 第2步：NameNode返回元数据
 * ● NameNode接收到请求后，会执行检查：
 * ○ 文件是否存在？
 * ○ 客户端是否有读取权限？
 * ● 如果检查通过，NameNode不会立即传输文件数据，而是将组成该文件的所有数据块（Blocks）的列表，以及每个数据块对应的所有副本所在的DataNode地址（按照HDFS的机架感知拓扑排序）返回给客户端。
 * ● 注意：NameNode只返回数据块的位置信息，不传输任何实际数据。
 * 第3步：客户端创建输入流（FSDataInputStream）
 * ● 客户端获得这些信息后，会创建一个 FSDataInputStream 对象。这个对象封装了读取文件所需的全部信息，包括数据块列表和DataNode位置。
 * ● FSDataInputStream 会透明地帮客户端管理整个读取过程。
 * 第4步：客户端连接到DataNode读取数据
 * ● 客户端现在准备读取数据。它首先会从第一个数据块开始。
 * ● FSDataInputStream 会从NameNode提供的该数据块的DataNode列表中，挑选一个在网络拓扑上距离客户端最近（例如，同一机架内）的DataNode建立连接，直接读取数据。
 * ● 选择最近的节点是为了最小化网络传输延迟，提高读取性能。
 * 第5步：数据流式读取
 * ● 客户端开始从选定的DataNode流式读取数据。
 * ● 数据从DataNode源源不断地通过网络传输到客户端。
 * ● 客户端读取的数据会先在本地进行缓存，然后由应用程序从流中读取。
 * 第6步：读取完毕，关闭连接
 * ● 当当前这个数据块的所有数据都读取完毕后，FSDataInputStream 会关闭与这个DataNode的连接。
 * 第7步：读取下一个数据块
 * ● 客户端然后开始读取下一个数据块。
 * ● 重复步骤4-6：FSDataInputStream 会再次为下一个数据块选择最合适的DataNode，建立连接并读取数据。
 * ● 这个过程循环往复，直到文件中的所有数据块都被读取完毕。
 * 第8步：客户端关闭输入流
 * ● 当所有数据都读取完成后，客户端调用 FSDataInputStream 的 close() 方法，关闭输入流，释放资源。
 * </p>
 *
 * @author Cyber
 * <p> Created By 2025/9/9
 * @version 1.0
 */
@Slf4j
public class HdfsReadTest {
    public static void main(String[] args) throws IOException {
        log.info("开始读取文件...");
        // 设置客户端用户名：lifangyu具备在hdfs读写权限
        System.setProperty("HADOOP_USER_NAME", "lifangyu");
        // 创建conf对象
        Configuration conf = new Configuration();
        // 设置操作的文件系统是DFS，默认是本地文件系统file:///
        conf.set("fs.defaultFS", "hdfs://master:9000");
        conf.set("dfs.replication", "1");
        // 创建FileSystem对象，其是文件系统操作的入口
        FileSystem fs = FileSystem.get(conf);
        log.info("FileSystem [{}] success!", fs.getClass().getName());
        // 调用open方法读取文件【读数据的核心入口】
        try (FSDataInputStream in = fs.open(new Path("/test/test.txt"));
             // 创建本地文件输出流
             FileOutputStream out = new FileOutputStream("/Users/lifangyu/Documents/project/bigdata/bigdata-project/hadoop/src/main/resources/files/test_read.txt")
        ) {
            log.info("FSDataInputStream open read data success!");
            // IO工具类实现流拷贝
            org.apache.commons.io.IOUtils.copy(in, out);
            log.info("IOUtils copy data success!");
        }
        // 关闭连接
        fs.close();
        log.info("读取文件完成!");
    }
}
